package com.flink.hbase.windowingoptimization.watermark;

import com.flink.hbase.windowingoptimization.config.WatermarkConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Properties;

/**
 * 实时用户行为分析示例
 * 展示Watermark在实时分析场景中的应用
 * 
 * 业务场景：
 * 1. 用户点击流实时统计
 * 2. 用户会话分析
 * 3. 实时推荐系统
 * 4. 异常行为检测
 * 
 * @author Flink Developer
 */
public class RealTimeAnalyticsExample {
    
    private static final Logger logger = LoggerFactory.getLogger(RealTimeAnalyticsExample.class);
    
    /**
     * 用户行为事件
     */
    public static class UserBehaviorEvent {
        public String userId;
        public String eventType;  // click, view, purchase, etc.
        public String productId;
        public String sessionId;
        public long timestamp;
        public String deviceType;
        public String location;
        public double value;
        
        public UserBehaviorEvent() {}
        
        public UserBehaviorEvent(String userId, String eventType, String productId, 
                               String sessionId, long timestamp, String deviceType, 
                               String location, double value) {
            this.userId = userId;
            this.eventType = eventType;
            this.productId = productId;
            this.sessionId = sessionId;
            this.timestamp = timestamp;
            this.deviceType = deviceType;
            this.location = location;
            this.value = value;
        }
        
        @Override
        public String toString() {
            return String.format("UserBehaviorEvent{userId='%s', eventType='%s', productId='%s', " +
                    "sessionId='%s', timestamp=%d, deviceType='%s', location='%s', value=%.2f}",
                    userId, eventType, productId, sessionId, timestamp, deviceType, location, value);
        }
    }
    
    /**
     * 用户行为统计结果
     */
    public static class UserBehaviorStats {
        public String userId;
        public String eventType;
        public long windowStart;
        public long windowEnd;
        public long eventCount;
        public double totalValue;
        public double avgValue;
        public String topProduct;
        public int uniqueProducts;
        
        public UserBehaviorStats() {}
        
        public UserBehaviorStats(String userId, String eventType, long windowStart, 
                               long windowEnd, long eventCount, double totalValue, 
                               double avgValue, String topProduct, int uniqueProducts) {
            this.userId = userId;
            this.eventType = eventType;
            this.windowStart = windowStart;
            this.windowEnd = windowEnd;
            this.eventCount = eventCount;
            this.totalValue = totalValue;
            this.avgValue = avgValue;
            this.topProduct = topProduct;
            this.uniqueProducts = uniqueProducts;
        }
        
        @Override
        public String toString() {
            return String.format("UserBehaviorStats{userId='%s', eventType='%s', " +
                    "window=[%s, %s], eventCount=%d, totalValue=%.2f, avgValue=%.2f, " +
                    "topProduct='%s', uniqueProducts=%d}",
                    userId, eventType, 
                    formatTimestamp(windowStart), formatTimestamp(windowEnd),
                    eventCount, totalValue, avgValue, topProduct, uniqueProducts);
        }
        
        private String formatTimestamp(long timestamp) {
            return LocalDateTime.ofEpochSecond(timestamp / 1000, 0, 
                java.time.ZoneOffset.UTC).format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
        }
    }
    
    /**
     * 用户行为聚合函数
     */
    public static class UserBehaviorAggregateFunction 
            implements AggregateFunction<UserBehaviorEvent, UserBehaviorAccumulator, UserBehaviorStats> {
        
        @Override
        public UserBehaviorAccumulator createAccumulator() {
            return new UserBehaviorAccumulator();
        }
        
        @Override
        public UserBehaviorAccumulator add(UserBehaviorEvent event, UserBehaviorAccumulator acc) {
            acc.eventCount++;
            acc.totalValue += event.value;
            acc.products.put(event.productId, acc.products.getOrDefault(event.productId, 0) + 1);
            
            // 更新最受欢迎的产品
            if (acc.topProduct == null || 
                acc.products.get(event.productId) > acc.products.getOrDefault(acc.topProduct, 0)) {
                acc.topProduct = event.productId;
            }
            
            return acc;
        }
        
        @Override
        public UserBehaviorStats getResult(UserBehaviorAccumulator acc) {
            return new UserBehaviorStats(
                acc.userId,
                acc.eventType,
                acc.windowStart,
                acc.windowEnd,
                acc.eventCount,
                acc.totalValue,
                acc.eventCount > 0 ? acc.totalValue / acc.eventCount : 0,
                acc.topProduct,
                acc.products.size()
            );
        }
        
        @Override
        public UserBehaviorAccumulator merge(UserBehaviorAccumulator acc1, UserBehaviorAccumulator acc2) {
            acc1.eventCount += acc2.eventCount;
            acc1.totalValue += acc2.totalValue;
            
            // 合并产品统计
            acc2.products.forEach((product, count) -> 
                acc1.products.merge(product, count, Integer::sum));
            
            // 重新计算最受欢迎的产品
            acc1.topProduct = acc1.products.entrySet().stream()
                .max(java.util.Map.Entry.comparingByValue())
                .map(java.util.Map.Entry::getKey)
                .orElse(null);
            
            return acc1;
        }
    }
    
    /**
     * 用户行为聚合器
     */
    public static class UserBehaviorAccumulator {
        public String userId;
        public String eventType;
        public long windowStart;
        public long windowEnd;
        public long eventCount = 0;
        public double totalValue = 0;
        public String topProduct;
        public java.util.Map<String, Integer> products = new java.util.HashMap<>();
    }
    
    /**
     * 窗口处理函数
     */
    public static class UserBehaviorProcessWindowFunction 
            extends ProcessWindowFunction<UserBehaviorStats, UserBehaviorStats, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<UserBehaviorStats> elements, 
                          Collector<UserBehaviorStats> out) throws Exception {
            
            UserBehaviorStats stats = elements.iterator().next();
            
            // 设置窗口时间信息
            stats.windowStart = context.window().getStart();
            stats.windowEnd = context.window().getEnd();
            
            // 获取用户ID和事件类型
            String[] keyParts = key.split(":");
            stats.userId = keyParts[0];
            stats.eventType = keyParts[1];
            
            // 计算窗口延迟
            long currentTime = context.currentProcessingTime();
            long windowDelay = currentTime - context.window().getEnd();
            
            if (windowDelay > 10000) {  // 超过10秒
                logger.warn("窗口处理延迟过高: {}ms, 用户: {}, 事件类型: {}, 窗口: [{}, {}]",
                    windowDelay, stats.userId, stats.eventType, 
                    stats.windowStart, stats.windowEnd);
            }
            
            out.collect(stats);
        }
    }
    
    /**
     * 实时用户行为分析主流程
     */
    public void runRealTimeAnalytics() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        // 配置Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "user-behavior-analysis");
        kafkaProps.setProperty("auto.offset.reset", "latest");
        
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
            "user-behavior-events",
            new SimpleStringSchema(),
            kafkaProps
        );
        
        // 数据源
        DataStream<String> rawStream = env.addSource(kafkaConsumer);
        
        // 解析用户行为事件
        SingleOutputStreamOperator<UserBehaviorEvent> eventStream = rawStream
            .map(new UserBehaviorEventParser())
            .filter(event -> event != null)
            .name("parse-user-behavior-events");
        
        // 设置Watermark策略 - 低延迟场景
        SingleOutputStreamOperator<UserBehaviorEvent> watermarkedStream = eventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.lowLatencyStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            )
            .name("assign-watermarks");
        
        // 按用户ID和事件类型分组，5分钟滚动窗口统计
        SingleOutputStreamOperator<UserBehaviorStats> userBehaviorStats = watermarkedStream
            .keyBy(event -> event.userId + ":" + event.eventType)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(
                new UserBehaviorAggregateFunction(),
                new UserBehaviorProcessWindowFunction()
            )
            .name("user-behavior-stats");
        
        // 异常行为检测
        SingleOutputStreamOperator<String> anomalyDetection = userBehaviorStats
            .filter(stats -> detectAnomalies(stats))
            .map(stats -> "异常行为检测: " + stats.toString())
            .name("anomaly-detection");
        
        // 实时推荐
        SingleOutputStreamOperator<String> recommendations = userBehaviorStats
            .filter(stats -> "click".equals(stats.eventType) || "view".equals(stats.eventType))
            .keyBy(stats -> stats.userId)
            .map(new RecommendationGenerator())
            .name("real-time-recommendations");
        
        // 输出结果
        userBehaviorStats.print("用户行为统计");
        anomalyDetection.print("异常检测");
        recommendations.print("实时推荐");
        
        env.execute("Real-Time User Behavior Analytics");
    }
    
    /**
     * 用户行为事件解析器
     */
    public static class UserBehaviorEventParser implements MapFunction<String, UserBehaviorEvent> {
        
        @Override
        public UserBehaviorEvent map(String value) throws Exception {
            try {
                // 解析JSON格式的用户行为事件
                // 这里简化处理，实际应使用JSON解析库
                String[] parts = value.split(",");
                if (parts.length >= 8) {
                    return new UserBehaviorEvent(
                        parts[0].trim(),  // userId
                        parts[1].trim(),  // eventType
                        parts[2].trim(),  // productId
                        parts[3].trim(),  // sessionId
                        Long.parseLong(parts[4].trim()),  // timestamp
                        parts[5].trim(),  // deviceType
                        parts[6].trim(),  // location
                        Double.parseDouble(parts[7].trim())  // value
                    );
                }
            } catch (Exception e) {
                logger.error("解析用户行为事件失败: {}", value, e);
            }
            return null;
        }
    }
    
    /**
     * 异常行为检测
     */
    private boolean detectAnomalies(UserBehaviorStats stats) {
        // 检测异常行为的规则
        
        // 1. 点击频率异常（5分钟内超过1000次点击）
        if ("click".equals(stats.eventType) && stats.eventCount > 1000) {
            logger.warn("检测到异常点击行为: 用户 {} 在5分钟内点击 {} 次", 
                stats.userId, stats.eventCount);
            return true;
        }
        
        // 2. 购买金额异常（5分钟内购买金额超过10000）
        if ("purchase".equals(stats.eventType) && stats.totalValue > 10000) {
            logger.warn("检测到异常购买行为: 用户 {} 在5分钟内购买金额 {}", 
                stats.userId, stats.totalValue);
            return true;
        }
        
        // 3. 浏览产品数量异常（5分钟内浏览超过100个不同产品）
        if ("view".equals(stats.eventType) && stats.uniqueProducts > 100) {
            logger.warn("检测到异常浏览行为: 用户 {} 在5分钟内浏览 {} 个不同产品", 
                stats.userId, stats.uniqueProducts);
            return true;
        }
        
        return false;
    }
    
    /**
     * 实时推荐生成器
     */
    public static class RecommendationGenerator implements MapFunction<UserBehaviorStats, String> {
        
        @Override
        public String map(UserBehaviorStats stats) throws Exception {
            // 基于用户行为生成实时推荐
            StringBuilder recommendation = new StringBuilder();
            recommendation.append("用户 ").append(stats.userId).append(" 的实时推荐: ");
            
            if (stats.topProduct != null) {
                recommendation.append("基于热门产品 ").append(stats.topProduct).append(" 的相关推荐");
            }
            
            if (stats.eventCount > 50) {
                recommendation.append("，用户活跃度高，推荐个性化内容");
            }
            
            if (stats.avgValue > 100) {
                recommendation.append("，用户价值高，推荐高价值产品");
            }
            
            return recommendation.toString();
        }
    }
    
    /**
     * 会话分析示例
     */
    public void runSessionAnalysis() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        // 配置Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.setProperty("bootstrap.servers", "localhost:9092");
        kafkaProps.setProperty("group.id", "session-analysis");
        
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(
            "user-session-events",
            new SimpleStringSchema(),
            kafkaProps
        );
        
        DataStream<String> rawStream = env.addSource(kafkaConsumer);
        
        // 解析会话事件
        SingleOutputStreamOperator<UserBehaviorEvent> sessionEventStream = rawStream
            .map(new UserBehaviorEventParser())
            .filter(event -> event != null && event.sessionId != null)
            .name("parse-session-events");
        
        // 设置Watermark策略 - 会话分析需要更长的延迟容忍度
        SingleOutputStreamOperator<UserBehaviorEvent> watermarkedSessionStream = sessionEventStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.mixedScenarioStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            )
            .name("assign-session-watermarks");
        
        // 按会话ID分组，使用会话窗口
        SingleOutputStreamOperator<Tuple3<String, Long, Long>> sessionStats = watermarkedSessionStream
            .keyBy(event -> event.sessionId)
            .window(org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows.withGap(Time.minutes(30)))
            .aggregate(new SessionAggregateFunction())
            .name("session-statistics");
        
        // 输出会话统计
        sessionStats.print("会话统计");
        
        env.execute("Session Analysis");
    }
    
    /**
     * 会话聚合函数
     */
    public static class SessionAggregateFunction 
            implements AggregateFunction<UserBehaviorEvent, SessionAccumulator, Tuple3<String, Long, Long>> {
        
        @Override
        public SessionAccumulator createAccumulator() {
            return new SessionAccumulator();
        }
        
        @Override
        public SessionAccumulator add(UserBehaviorEvent event, SessionAccumulator acc) {
            if (acc.sessionId == null) {
                acc.sessionId = event.sessionId;
                acc.userId = event.userId;
                acc.startTime = event.timestamp;
                acc.endTime = event.timestamp;
            } else {
                acc.endTime = Math.max(acc.endTime, event.timestamp);
            }
            acc.eventCount++;
            return acc;
        }
        
        @Override
        public Tuple3<String, Long, Long> getResult(SessionAccumulator acc) {
            long sessionDuration = acc.endTime - acc.startTime;
            return new Tuple3<>(acc.sessionId, sessionDuration, acc.eventCount);
        }
        
        @Override
        public SessionAccumulator merge(SessionAccumulator acc1, SessionAccumulator acc2) {
            acc1.eventCount += acc2.eventCount;
            acc1.startTime = Math.min(acc1.startTime, acc2.startTime);
            acc1.endTime = Math.max(acc1.endTime, acc2.endTime);
            return acc1;
        }
    }
    
    /**
     * 会话累加器
     */
    public static class SessionAccumulator {
        public String sessionId;
        public String userId;
        public long startTime;
        public long endTime;
        public long eventCount = 0;
    }
    
    public static void main(String[] args) throws Exception {
        RealTimeAnalyticsExample example = new RealTimeAnalyticsExample();
        
        // 运行实时用户行为分析
        example.runRealTimeAnalytics();
        
        // 或者运行会话分析
        // example.runSessionAnalysis();
    }
} 